package com.group1.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.bean.TableProcess;
import com.group1.realtime.util.JdbcUtil;
import com.group1.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;
import java.util.Set;


public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    //    private DruidDataSource dataSource;
    private Connection conn;
    private ValueState<String> sqlState;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
//        dataSource = DruidDSUtil.createDataSource();
//        conn = dataSource.getConnection();
        conn = JdbcUtil.getPhoenixConnection();
        //创建一个键控状态
        sqlState = getRuntimeContext().getState(new ValueStateDescriptor<String>("sqlState", String.class));

        jedis = RedisUtil.getRedisClient();
    }

    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();//连接使从连接池获取的,所以这里的close不是关闭，而是归还
        }
        if (jedis != null) {
            jedis.close();
        }
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {

        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        //把维度数据写到phoenix
        saveDimToPhoenix(data, tp);
        //删除redis中的缓存
        delCache(data, tp);
        //更新redis中的缓存
        //updateCache(data, tp);

    }

    private void updateCache(JSONObject data, TableProcess tp) {
        String key = tp.getSinkTable() + ":" + data.getString("id");
        if ("update".equals(data.getString("operate_type")) && jedis.exists(key)) {
            data.remove("operate_type");
            JSONObject outer = new JSONObject();

            Set<Map.Entry<String, Object>> set = data.entrySet();
            for (Map.Entry<String, Object> entry : set) {
                outer.put(entry.getKey().toUpperCase(), entry.getValue());
            }
            jedis.setex(key, 2*24*60*60, outer.toJSONString());
        }
    }


    private void delCache(JSONObject data, TableProcess tp) {

        if ("update".equals(data.getString("operate_type"))) {
            String key = tp.getSinkTable() + ":" + data.getString("id");
            jedis.del(key);
        }

    }

    private void saveDimToPhoenix(JSONObject data, TableProcess tp) throws java.io.IOException, SQLException {
        //如果状态值没有值，表示该key对应的value第一次进入，即对应的表第一次插入数据
        if (sqlState.value()==null) {
            StringBuilder sql = new StringBuilder();
            //拼接sql语句
            sql.append("upsert into ")
                    .append(tp.getSinkTable())
                    .append("(")
                    .append(tp.getSinkColumns())
                    .append(")values(")
                    .append(tp.getSinkColumns().replaceAll("[^,]+","?"))
                    .append(")");
            //将sql语句存入状态中
            sqlState.update(sql.toString());
        }
        //从状态中获取sql语句
        String sql = sqlState.value();
        System.out.println("插入语句: \n" + sql);
        //预处理
        PreparedStatement ps = conn.prepareStatement(sql);
        //从tp配置信息中获取表的所有列名
        String[] columns = tp.getSinkColumns().split(",");
        //遍历
        for (int i = 0; i < columns.length; i++) {
            //获取每一个列名
            String name = columns[i];
            //从数据中获取对应的值
            String v = data.getString(name);
            //给sql中的占位符赋值
            ps.setString(i + 1, v);
        }
        //执行
        ps.execute();
        conn.commit();
        ps.close();
    }
}